/*
    实现异步工作器
*/

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include "buffer.hpp"
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>

namespace log_sys{
    using Functor = std::function<void(Buffer&)>;
    enum class TypeAsync{
        SAFE_ASYNC,
        UNSAFE_ASYNC
    };

    class AsyncLooper{
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor& callback, TypeAsync looppertype)
            :_stop(false)
            ,_thread(std::thread(&AsyncLooper::threadEntry, this))
            ,_callback(callback)
            ,_loopertype(TypeAsync::SAFE_ASYNC)
        {}

        ~AsyncLooper()
        {
            stop();
        }

        void stop(){
            _stop = true;
            _concond.notify_all(); // 唤醒所有工作线程
            _thread.join();
        }

        void push(const char* data, size_t len)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量值：如果生产缓冲区中剩余空间大小大于数据长度，则可以添加数据
            if (_loopertype == TypeAsync::SAFE_ASYNC)
            {
                _procond.wait(lock, [&](){return _stop || _probuf.writeablesize() >= len;});  
            }
            
            _probuf.push(data, len);
            _concond.notify_one();
        }

    private:
        void threadEntry()
        {
            while(1)
            {
                // 下面多加了一个{}是因为每次缓冲区交换完毕之后，都需要对锁进行释放，
                // 然后进入非临界区，多个线程并行执行回调函数   
                // 1. 判断生产缓冲区有没有数据，有则交换，无则阻塞
                {       
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 只有在 _stop 被设置为退出(即被设置为 1) 且生产缓冲区已经没有数据时，才退出循环
                    // 如果不设置这个条件进行判断，可能会出现_stop 为0，但是生产缓冲区中还有数据，但是退出的情况
                    if(_stop && _probuf.empty()) {
                        break;
                    }
                    // _stop为 true，此时条件变量释放，
                    // 往下再运行一次，并且释放 lock 这个互斥锁
                    // 退出循环
                    _concond.wait(lock, [&](){return !_probuf.empty();});
                    _conbuf.swap(_probuf);
                    // 2. 唤醒生产者
                    if(_loopertype == TypeAsync::SAFE_ASYNC)
                        _procond.notify_all();
                }
                // 3. 被唤醒后，对消费缓冲区进行数据处理
                _callback(_conbuf);
                // 4. 初始化消费缓冲区
                _conbuf.reset();
            }
        }
        
    private:
        Functor _callback;
    private:
        TypeAsync _loopertype;
        std::atomic<bool> _stop;
        Buffer _probuf; // 生产缓冲区
        Buffer _conbuf; // 消费缓冲区
        std::mutex _mutex;
        std::condition_variable _procond;
        std::condition_variable _concond;
        std::thread _thread;
    };
}

#endif
